package com.github.cossbow.nsq;

import com.github.cossbow.nsq.exceptions.HttpException;
import com.github.cossbow.nsq.util.FutureUtil;
import com.github.cossbow.nsq.util.TagThreadFactory;
import com.github.cossbow.nsq.util.ThrowoutFunction;
import com.google.gson.FieldNamingPolicy;
import com.google.gson.Gson;
import com.google.gson.GsonBuilder;
import io.netty.buffer.ByteBufInputStream;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import reactor.core.publisher.Mono;
import reactor.netty.http.client.HttpClient;

import java.io.IOException;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.Collection;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Function;

final
public class NSQUtil {
    private NSQUtil() {
    }


    public static final ScheduledExecutorService SCHEDULER
            = Executors.newSingleThreadScheduledExecutor(
            new TagThreadFactory("nsq-schedule"));

    public final static ExecutorService EXECUTOR
            = Executors.newCachedThreadPool(
            new TagThreadFactory("nsq-c"));

    private static final HttpClient CLIENT;

    private static final Gson GSON;

    static {
        CLIENT = HttpClient.create().doOnResponse((response, connection) -> {
            var status = response.status().code();
            if (200 != status) {
                throw new HttpException("HTTP " + status);
            }
        });

        GSON = new GsonBuilder()
                .setFieldNamingPolicy(FieldNamingPolicy.LOWER_CASE_WITH_UNDERSCORES)
                .create();
    }

    public static <T> String toJson(T v) {
        return GSON.toJson(v);
    }

    public static <T> T fromJson(String src, Class<T> type) {
        return GSON.fromJson(src, type);
    }

    public static <T> T fromJson(InputStream src, Class<T> type) {
        try (var reader = new InputStreamReader(src)) {
            return GSON.fromJson(reader, type);
        } catch (IOException e) {
            throw new IllegalArgumentException("invalid json data", e);
        }
    }


    public static <T> Mono<T> get(String uri, Class<T> type) {
        return CLIENT.get().uri(uri).responseContent().aggregate()
                .asInputStream().map(in -> fromJson(in, type));
    }

    public static Mono<String> post(String uri) {
        return CLIENT.post().uri(uri).responseContent()
                .aggregate().asString(StandardCharsets.UTF_8);
    }

    //

    public static ServerAddress parseAddress(String nsqdAddress) {
        var a = nsqdAddress.split(":");
        if (a.length != 2) {
            throw new IllegalArgumentException("illegal address: " + nsqdAddress);
        }
        return new ServerAddress(a[0], Integer.parseInt(a[1]));
    }

    //

    public static ThrowoutFunction<ByteBufInputStream, String, IOException> stringDecoder() {
        return in -> {
            var reader = new InputStreamReader(in);
            var sb = new StringBuilder();
            var buf = new char[1024];
            int nRead;
            while ((nRead = reader.read(buf)) != -1) {
                sb.append(buf, 0, nRead);
            }
            return sb.toString();
        };
    }

    //

    public
    static CompletableFuture<Void>
    unSubscribe(Collection<NSQConsumer<?>> list) {
        if (null == list || list.isEmpty()) {
            return CompletableFuture.completedFuture(null);
        }
        if (1 == list.size()) {
            return list.iterator().next().unSubscribeAsync();
        }

        var futures = new ArrayList<CompletableFuture<Void>>();
        for (var consumer : list) {
            futures.add(consumer.unSubscribeAsync());
        }
        return FutureUtil.allOf(futures);
    }

    //

    public static <T> CompletableFuture<T> future(
            ChannelFuture cf,
            Function<Channel, T> call,
            Executor executor) {
        try {
            var future = new CompletableFuture<T>();
            cf.addListener(f -> {
                if (f.isSuccess()) {
                    var ch = cf.channel();
                    future.completeAsync(() -> call.apply(ch),
                            null == executor ? future.defaultExecutor() : executor);
                } else {
                    future.completeExceptionally(f.cause());
                }
            });
            return future;
        } catch (Throwable e) {
            return CompletableFuture.failedFuture(e);
        }
    }

    public static <T> CompletableFuture<T> future(
            ChannelFuture cf,
            Function<Channel, T> call) {
        return future(cf, call, null);
    }

    public static CompletableFuture<Channel> future(
            ChannelFuture cf) {
        return future(cf, Function.identity());
    }

}
